package edu.vt.ece.rt.benchmark.intruder;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;

import edu.vt.ece.rt.benchmark.BenchmarkThread;
import edu.vt.ece.rt.benchmark.hashmap.Hashmap;
import edu.vt.ece.rt.benchmark.intruder.Error.error;
import edu.vt.ece.rt.dstm.Context;
import edu.vt.ece.rt.dstm.ContextDelegator;
import edu.vt.ece.rt.dstm.QObject;
import edu.vt.ece.rt.dstm.TMProxy;
import edu.vt.ece.rt.quorum.ClustersManager;
import edu.vt.ece.rt.quorum.Resource;
import edu.vt.ece.rt.quorum.util.AbortException;
import edu.vt.ece.rt.util.Logger.Logger;


public class Decoder extends QObject{


	HashMap<Integer,ArrayList<Packet>> fragmentedMapPtr;  /* contains list of Packet* */
	Queue<Decoded> decodedQueuePtr; /* contains decoded_t* */
	String Id;

	public Decoder (String id) {
		fragmentedMapPtr = new HashMap<Integer,ArrayList<Packet>>();
		//	        assert(decoderPtr.fragmentedMapPtr);
		decodedQueuePtr = new LinkedList<Decoded>();
		//	        assert(decoderPtr.decodedQueuePtr);
		Id = id;
		Resource.putQObject(getId(), this);
	}

	public Decoder () {
		fragmentedMapPtr = new HashMap<Integer,ArrayList<Packet>>();
		//	        assert(decoderPtr.fragmentedMapPtr);
		decodedQueuePtr = new LinkedList<Decoded>();
		//	        assert(decoderPtr.decodedQueuePtr);
	}

	/* =============================================================================
	 * decoder_process
	 * =============================================================================
	 */
	static Error.error
	decoder_process (Packet p, long numByte)
	{
		boolean status;

		/*
		 * Basic error checking
		 */

		if (numByte < Packet.PACKET_HEADER_LENGTH) {
			return Error.error.ERROR_SHORT;
		}

		Packet packetPtr = p;
		Integer flowId      = packetPtr.flowId;
		long fragmentId  = packetPtr.fragmentId;
		long numFragment = packetPtr.numFragment;
		long length      = packetPtr.length;

		if (flowId < 0) {
			return Error.error.ERROR_FLOWID;
		}

		if ((fragmentId < 0) || (fragmentId >= numFragment)) {
			return Error.error.ERROR_FRAGMENTID;
		}

		if (length < 0) {
			return Error.error.ERROR_LENGTH;
		}

		/*
		 * With the above checks, this one is redundant
		 */
		 if (numFragment < 1) {
			 return Error.error.ERROR_NUMFRAGMENT;
		 }

		 /*
		  * Add to fragmented map for reassembling
		  */

		 Decoded decodedPtr = null;
		 if (numFragment > 1) {

			 decodedPtr = processFragmentList(packetPtr, flowId, null);
			 
		 } else {

			 /*
			  * This is the only fragment, so it is ready
			  */

//			 System.out.println("Fragment of length 1");
			 if (fragmentId != 0) {
				 return Error.error.ERROR_FRAGMENTID;
			 }

			 String data = packetPtr.data;
			 decodedPtr = new Decoded();
			 decodedPtr.flowId = flowId;
			 decodedPtr.data = data;
			 addDecodedQueue(decodedPtr,null);
//			 Queue<Decoded> decodedQueuePtr = decoderPtr.decodedQueuePtr;
//			 decodedQueuePtr.add(decodedPtr);
		 }

		 return Error.error.ERROR_NONE;
	}

	
	public static void addDecodedQueue(Decoded decodedPtr, Context parentContext) {
		
		Long priority = System.currentTimeMillis();

		for (int i = 0; i < 0x7fffffff; i++) {

			((BenchmarkThread)Thread.currentThread()).collectStatsRoot(i);
			Context context = ContextDelegator.getNestedContext();
			context.setParent(parentContext);
			context.setDescription("add decoded to queue");
			context.sanityNested();
			context.setPriority(priority);

			Logger.debug("----Started----- R Transaction "+context.getContextId()+" on "+ClustersManager.baseClusterId);
			try {
				
				
				Decoder d = (Decoder) TMProxy.getTMProxy().write("decoder", context.getContextId());
				d.decodedQueuePtr.add(decodedPtr);
				
				if (TMProxy.getTMProxy().requestCommit(context.getContextId())) {
					context.unsetParent(parentContext);
					return;
				}
			} catch (AbortException e) {
				Logger.debug("Aborting in loop "+context.getContextId());
				e.touch();
			}
		}
		Logger.fetal("Unable to complete transaction in given number of tries");

		return;
	}

	public static Decoded getDecodedPtr(Context parentContext) {
		
		Long priority = System.currentTimeMillis();

		for (int i = 0; i < 0x7fffffff; i++) {

			((BenchmarkThread)Thread.currentThread()).collectStatsRoot(i);
			Context context = ContextDelegator.getNestedContext();
			context.setParent(parentContext);
			context.setDescription("add decoded to queue");
			context.sanityNested();
			context.setPriority(priority);

			Logger.debug("----Started----- R Transaction "+context.getContextId()+" on "+ClustersManager.baseClusterId);
			try {
				
				
				Decoder d = (Decoder) TMProxy.getTMProxy().write("decoder", context.getContextId());
				Decoded decoded = d.decodedQueuePtr.poll();
				
				if (TMProxy.getTMProxy().requestCommit(context.getContextId())) {
					context.unsetParent(parentContext);
					return decoded;
				}
			} catch (AbortException e) {
				Logger.debug("Aborting in loop "+context.getContextId());
				e.touch();
			}
		}
		Logger.fetal("Unable to complete transaction in given number of tries");

		return null;
	}

	/* =============================================================================
	 * decoder_getComplete
	 * -- If none, returns NULL
	 * =============================================================================
	 */

	static Pair
	decoder_getComplete (Integer decodedFlowIdPtr)
	{
		
		String data;
		Decoded decodedPtr = getDecodedPtr(null);
		

		Pair p = new Pair();

		if (decodedPtr != null) {
			decodedFlowIdPtr = decodedPtr.flowId;
			data = decodedPtr.data;
			//	            free(decodedPtr);
			p.str = data;
			p.i = decodedFlowIdPtr;
		} else {
			decodedFlowIdPtr = -1;
			data = null;
			p.str = data;
			p.i = decodedFlowIdPtr;
		}

		return p;
	}


	public static Decoded processFragmentList(Packet packetPtr, long flowId,
										Context parentContext) {

		Long priority = System.currentTimeMillis();
		String fragmentListId = flowId+"-fragment";
		
		for (int i = 0; i < 0x7fffffff; i++) {

			((BenchmarkThread)Thread.currentThread()).collectStatsRoot(i);
			Context context = ContextDelegator.getNestedContext();
			context.setParent(parentContext);
			context.setDescription("find element in bucket");
			context.sanityNested();
			context.setPriority(priority);

			Logger.debug("----Started----- R Transaction "+context.getContextId()+" on "+ClustersManager.baseClusterId);
			try {

				Decoded decodedPtr = null;
				 FragmentList f = (FragmentList)TMProxy.getTMProxy().write(fragmentListId, context.getContextId());
				 
				 if (f.fragmentListPtr == null) {

					 f.fragmentListPtr = new ArrayList<Packet>();
					 assert(f.fragmentListPtr != null);
					 f.fragmentListPtr.add(packetPtr);
				 } else {

					 Iterator<Packet> it;
					 it = f.fragmentListPtr.iterator();
					 Packet firstFragmentPtr = it.next();
					 long expectedNumFragment = firstFragmentPtr.numFragment;
					 long numFragment = packetPtr.numFragment;
					 
					 if (numFragment != expectedNumFragment) {
						 return null;
					 }

					 f.fragmentListPtr.add(packetPtr);
					 Collections.sort(f.fragmentListPtr);
					 
					 /*
					  * If we have all the fragments we can reassemble them
					  */
					 if (f.fragmentListPtr.size() == numFragment) {
						 long numByte = 0;
						 long i1 = 0;
						 it = f.fragmentListPtr.iterator();
						 while (it.hasNext()) {
							 Packet fragmentPtr = it.next();
							 assert(fragmentPtr.flowId == flowId);
							 if (fragmentPtr.fragmentId != i1) {
								 return null; /* should be sequential */
							 }
							 numByte += fragmentPtr.length;
							 i1++;
						 }

						 String data = new String();
						 String dst = data;
						 it = f.fragmentListPtr.iterator();
						 while (it.hasNext()) {
							 Packet fragmentPtr =
									 it.next();
							 dst = dst + fragmentPtr.data;
						 }

						 decodedPtr = new Decoded();
						 decodedPtr.flowId = (int)flowId;
						 decodedPtr.data = dst;
//
//						 Queue<Decoded> decodedQueuePtr = decoderPtr.decodedQueuePtr;
//						 decodedQueuePtr.add(decodedPtr);

//						 addDecodedQueue(decodedPtr, null);
						Decoder d = (Decoder) TMProxy.getTMProxy().write("decoder", context.getContextId());
						d.decodedQueuePtr.add(decodedPtr);
						 
					 }

				 }
				 
				if (TMProxy.getTMProxy().requestCommit(context.getContextId())) {
					context.unsetParent(parentContext);
					return decodedPtr;
				}
				
			} catch (AbortException e) {
				Logger.debug("Aborting in loop "+context.getContextId());
				e.touch();
			}
		}
		
		return  null;

	}


		/**
		 * @param args
		 */
		public static void main(String[] args) {
			Decoder decoderPtr;

			System.out.println("Starting...");

			decoderPtr = new Decoder();
			//        assert(decoderPtr);

			long numDataByte = 3;
			long numPacketByte = Packet.PACKET_HEADER_LENGTH + numDataByte;

			String abcBytes;
			//        assert(abcBytes);
			Packet abcPacketPtr = new Packet();
			//        abcPacketPtr = (Packet)abcBytes;
			abcPacketPtr.flowId = 1;
			abcPacketPtr.fragmentId = 0;
			abcPacketPtr.numFragment = 2;
			abcPacketPtr.length = numDataByte;
			abcPacketPtr.data = "abc";

			String defBytes ;
			//        assert(defBytes);
			Packet defPacketPtr = new Packet();
			//        defPacketPtr = (Packet)defBytes;
			defPacketPtr.flowId = 1;
			defPacketPtr.fragmentId = 1;
			defPacketPtr.numFragment = 2;
			defPacketPtr.length = numDataByte;
			defPacketPtr.data = "def";

			assert(decoder_process(abcPacketPtr, numDataByte) == Error.error.ERROR_SHORT);

			abcPacketPtr.flowId = -1;
			assert(decoder_process(abcPacketPtr, numPacketByte) == Error.error.ERROR_FLOWID);
			abcPacketPtr.flowId = 1;
			//
			abcPacketPtr.fragmentId = -1;
			assert(decoder_process(abcPacketPtr, numPacketByte) == Error.error.ERROR_FRAGMENTID);
			abcPacketPtr.fragmentId = 0;
			//
			abcPacketPtr.fragmentId = 2;
			assert(decoder_process( abcPacketPtr, numPacketByte) == Error.error.ERROR_FRAGMENTID);
			abcPacketPtr.fragmentId = 0;
			//
			abcPacketPtr.fragmentId = 2;
			assert(decoder_process(abcPacketPtr, numPacketByte) == Error.error.ERROR_FRAGMENTID);
			abcPacketPtr.fragmentId = 0;
			//
			abcPacketPtr.length = -1;
			assert(decoder_process( abcPacketPtr, numPacketByte) == Error.error.ERROR_LENGTH);
			abcPacketPtr.length = numDataByte;
			//
			assert(decoder_process(abcPacketPtr, numPacketByte) == Error.error.ERROR_NONE);
			defPacketPtr.numFragment = 3;
			assert(decoder_process( defPacketPtr, numPacketByte) == Error.error.ERROR_NUMFRAGMENT);
			defPacketPtr.numFragment = 2;
			//
			assert(decoder_process(abcPacketPtr, numPacketByte) == Error.error.ERROR_NONE);
			defPacketPtr.fragmentId = 0;
			assert(decoder_process( defPacketPtr, numPacketByte) == Error.error.ERROR_INCOMPLETE);
			defPacketPtr.fragmentId = 1;

			Integer flowId = new Integer(1);
			assert(decoder_process( defPacketPtr, numPacketByte) == Error.error.ERROR_NONE);
			assert(decoder_process(abcPacketPtr, numPacketByte) == Error.error.ERROR_NONE);
			//        String str = decoder_getComplete(decoderPtr, flowId);
			//        assert(str.equals("abcdef"));
			////        free(str);
			//        assert(flowId == 1);
			//
			abcPacketPtr.numFragment = 1;
			assert(decoder_process( abcPacketPtr, numPacketByte) == Error.error.ERROR_NONE);
			//        str = decoder_getComplete(decoderPtr, flowId);
			//        assert(str.equals("abc"));
			////        free(str);
			abcPacketPtr.numFragment = 2;
			assert(flowId == 1);
			//
			//        str = decoder_getComplete(decoderPtr, flowId);
			//        assert(str == null);
			//        assert(flowId == -1);

			//        decoder_free(decoderPtr);
			//
			//        free(abcBytes);
			//        free(defBytes);

			System.out.println("All tests passed.");

			//        return 0;

		}

		@Override
		public QObject clone() {
			Decoder d = new Decoder();
			d.setCloneOwner(getCloneOwner());
			d.setVersion(getVersion());
			d.Id = getId();

			Iterator<Decoded> it = decodedQueuePtr.iterator();
			while (it.hasNext()) {
				Decoded item = it.next();
				d.decodedQueuePtr.add(item);
			}
//			d.decodedQueuePtr = new LinkedList<Decoded>(decodedQueuePtr); 
			return d;
		}

		@Override
		public String getId() {
			// TODO Auto-generated method stub
			return Id;
		}

	}
